package com.cl.spark.node;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

@Component
public class FilterDateNode extends BaseSparkNode {
    @Autowired
    SparkSession sparkSession;

    @Override
    public SparkResult process(SparkParam sparkParam) {
        Dataset<Row> dataset = sparkParam.getSparkResultList().get(0).getDataset();
        JSONObject expression = sparkParam.getNodeExpression();

        JSONArray fields = expression.getJSONArray("fields");
        JSONArray fields2 = expression.getJSONArray("fields2");
        String field = fields.getString(0);
        String filterFunction = expression.getString("filterFunction");
        String filterValue = expression.getString("filterValue");

        Column generateFilterColumn;
        if (fields2 != null && !fields2.isEmpty()) {
            String field2 = fields2.getString(0);
            generateFilterColumn = generateFilterColumn2(field, filterFunction, field2);
        } else {
            generateFilterColumn = generateFilterColumn(field, filterFunction, filterValue);
        }
        Dataset<Row> filter = dataset.filter(generateFilterColumn);
        return SparkResult.success(filter);

    }

    public Column generateFilterColumn(String field, String filterFunction, String filterValue) {
        if (!StringUtils.hasLength(field) || !StringUtils.hasLength(filterFunction)) {
            throw new RuntimeException(String.format("field: %s function %s value %s", field, filterFunction, filterValue));
        }
        Column column = functions.to_date(new Column(field), "yyyy-MM-dd");
        Column dateLit = functions.to_date(functions.lit(filterValue), "yyyy-MM-dd");
        if (filterValue.equals("今天")){
            dateLit = functions.current_date();
        }

        switch (filterFunction) {
            case "大于":
                column = column.gt(dateLit);
                break;
            case "小于":
                column = column.lt(dateLit);
                break;
            case "等于":
                column = column.equalTo(dateLit);
                break;
            case "不等于":
                column = column.notEqual(dateLit);
                break;
            case "大于等于":
                column = column.geq(dateLit);
                break;
            case "小于等于":
                column = column.leq(dateLit);
                break;
        }
        return column;
    }

    public Column generateFilterColumn2(String field, String filterFunction, String field2) {
        if (!StringUtils.hasLength(field) || !StringUtils.hasLength(filterFunction)) {
            throw new RuntimeException(String.format("field: %s function %s value %s", field, filterFunction, field2));
        }
        Column column = functions.to_date(new Column(field), "yyyy-MM-dd");
        Column dateLit = functions.to_date(new Column(field2), "yyyy-MM-dd");

        switch (filterFunction) {
            case "大于":
                column = column.gt(dateLit);
                break;
            case "小于":
                column = column.lt(dateLit);
                break;
            case "等于":
                column = column.equalTo(dateLit);
                break;
            case "不等于":
                column = column.notEqual(dateLit);
                break;
            case "大于等于":
                column = column.geq(dateLit);
                break;
            case "小于等于":
                column = column.leq(dateLit);
                break;
        }
        return column;
    }



    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.FILTER_DATE;
    }
}
